package com.sui.bigdata.flink.sql.function.aggregate;

import com.sui.bigdata.flink.sql.function.util.PhoneHHIAccum;

import org.apache.flink.table.functions.AggregateFunction;

import java.util.Iterator;

import static com.sui.bigdata.flink.sql.function.util.Utils.isMobileNO;

/**
 * @author LiuHuan
 * @date 2019/9/5 16:09
 * @description
 * @reviewer
 */
//通讯录号码总数
public class UserknPhoneAll extends AggregateFunction<Integer, PhoneHHIAccum> {
    @Override
    public PhoneHHIAccum createAccumulator() {
        return new PhoneHHIAccum();
    }

    @Override
    public Integer getValue(PhoneHHIAccum phoneHHIAccum) {
        Integer count = 0;
        if (!(phoneHHIAccum == null || phoneHHIAccum.getCount() == 0)) {
            count = phoneHHIAccum.getCount();
        }
        return count;
    }

    public void accumulate(PhoneHHIAccum phoneHHIAccum, String phone) {
        if (!phoneHHIAccum.exitsPhone(phone) && isMobileNO(phone)) {
            phoneHHIAccum.addphone(phone);
            phoneHHIAccum.setCount(phoneHHIAccum.getCount() + 1);
        }
    }

    public void merge(PhoneHHIAccum phoneHHIAccum, Iterable<PhoneHHIAccum> it) {
        Iterator<PhoneHHIAccum> iter = it.iterator();
        while (iter.hasNext()) {
            PhoneHHIAccum a = iter.next();
            phoneHHIAccum.setGroups(a.getGroups());
            phoneHHIAccum.setCount(a.getCount());
            phoneHHIAccum.setRepetition(a.getRepetition());
        }
    }
}
